package com.ecommerce.function;
import com.ecommerce.model.ProductData;import org.apache.flink.api.common.state.ListState;import org.apache.flink.api.common.state.ListStateDescriptor;import org.apache.flink.configuration.Configuration;import org.apache.flink.streaming.api.functions.KeyedProcessFunction;import org.apache.flink.util.Collector;
import java.util.ArrayList;import java.util.Comparator;import java.util.List;
public class GrowthRateRankingFunction extends KeyedProcessFunction<String, ProductData, ProductData> {
    
    private transient ListState<ProductData> productState;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        ListStateDescriptor<ProductData> descriptor = 
            new ListStateDescriptor<>("productState", ProductData.class);
        productState = getRuntimeContext().getListState(descriptor);
    }
    
    @Override
    public void processElement(ProductData product, Context ctx, Collector<ProductData> out) throws Exception {
        productState.add(product);
        long triggerTime = (System.currentTimeMillis() / 3600000 + 1) * 3600000;
        ctx.timerService().registerProcessingTimeTimer(triggerTime);
    }
    
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<ProductData> out) throws Exception {
        List<ProductData> products = new ArrayList<>();
        for (ProductData product : productState.get()) {
            products.add(product);
        }
        
        // 按增长率排序
        products.sort(Comparator.comparing(ProductData::getGrowthRate).reversed());
        
        int rank = 1;
        for (int i = 0; i < Math.min(100, products.size()); i++) {
            ProductData product = products.get(i);
            product.setRank(rank);
            product.setRankType("growth_rate");
            product.setTimestamp(System.currentTimeMillis());
            out.collect(product);
            rank++;
        }
        
        productState.clear();
    }}